package com.cl.process

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.alibaba.fastjson.JSON
import com.cl.bean.WaterQualityBean
import com.cl.util
import com.cl.util.BaseJdbcSink
import org.apache.spark.sql.Encoders
//import org.apache.hadoop.ipc.Server.Connection
import org.apache.spark.SparkContext
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.SparkConf
import org.apache.spark.sql.ForeachWriter





/**
 * @Author xc
 * @Date 2023/4/16 16:41
 * @Version 1.0
 */

object  SewageDischarge_Data_Process {
  def main(args: Array[String]): Unit = {


    //1.创建StructuredStreaming执行环境：直接创建SparkSession即可
    //StructuredStreaming支持使用SQL来处理实时流数据,数据抽象和SparkSQL一样,也是DataFrame和DataSet
    //所以这里创建StructuredStreamin执行环境就直接创建SparkSession即可
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SewageDischarge_Data_Process").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //导入隐式转换方便后续使用
    import org.apache.spark.sql.functions._
    import spark.implicits._

    import scala.collection.JavaConversions._

    //2.连接Kafka
    //从kafka接收消息
    val kafkaDF: DataFrame = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "192.168.32.129:9092")
      .option("subscribe", "sewageDischarge")
      .load()
    //取出消息中的value
    val jsonStrDS: Dataset[String] = kafkaDF.selectExpr("CAST(value AS STRING)").as[String]
//    输出收到来自kafka的json数据
//        jsonStrDS.writeStream
//          .format("console")//输出目的地
//          .outputMode("append")//输出模式,默认就是append表示显示新增行
//          .trigger(Trigger.ProcessingTime(0))//触发间隔,0表示尽可能快的执行
//          .option("truncate",false)//表示如果列名过长不进行截断
//          .start()
//          .awaitTermination()//等待 Streaming 应用程序停止


//    数据信息
    //    中文值
    //    |{"mid":1,"pH(无量纲)":7.74,"叶绿素α(mg/L)":0.0,"总氮(mg/L)":1.72,"总磷(mg/L)":0.06,"断面位点":"前江口",
    //    "氨氮(mg/L)":0.03,"水温(℃)":19.64,"水质 类别":"Ⅱ","流域河流":"长江流域 长江","浊度(NTU)":42.81,"溶解氧(mg/L)":8.13,"电导率(μS/cm)":297.59,
    //    "监测时间":"04-16 20点","省市":"安徽省 安庆市","站点 情况":"正常","藻密度(cells/L)":0.0,"高锰酸盐指数(mg/L)":1.96}                     |

    //  原属性值
    //  {"algalDensity":2144860.0,"ammoniaNitrogen":0.03,"chlorophyllAlpha":0.02,"conductivity":468.12,"dissolvedOxygen":9.51,"mid":1,"monitoringTime":"04-17 16点","pH":8.64,"permanganateIndex":3.45,"provinceCity":"安徽省 合肥市","sectionPoint":"兆河入湖区","siteCondition":"正常","totalNitrogen":1.42,"totalPhosphorus":0.03,"turbidity":27.31,"waterQualityType":"Ⅲ","waterTemperature":19.88,"watershedRiver":"巢湖流域 巢湖"}     |

    //    3.处理数据
val WaterQualities:Dataset[WaterQualityBean] = jsonStrDS.map(jsonStr => {
  //注意:Scala中获取class对象使用classOf[类名]
  //Java中使用类名.class/Class.forName(全类路径)/对象.getClass()
  JSON.parseObject(jsonStr, classOf[WaterQualityBean])
})
//    输出对应的实体类数据信息
//    WaterQualities.writeStream
//      .format("console")//输出目的地
//      .outputMode("append")//输出模式,默认就是append表示显示新增行
//      .trigger(Trigger.ProcessingTime(0))//触发间隔,0表示尽可能快的执行
//      .option("truncate",false)//表示如果列名过长不进行截断
//      .start()
//      .awaitTermination()

//    WaterQualities.writeStream
//      .format("console")
//      .outputMode("append")
//      .trigger(Trigger.ProcessingTime(0))
//      .option("truncate",false)
//      .start()

/*表结构
CREATE TABLE `water_quality`  (
  `Mid` int(11) NOT NULL AUTO_INCREMENT,
  `Water_Temperature` float NULL DEFAULT NULL,
  `Turbidity` float NULL DEFAULT NULL,
  `Chlorophyll_Alpha` float NULL DEFAULT NULL,
  `Watershed_River` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  `Province_City` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  `Dissolved_Oxygen` float NULL DEFAULT NULL,
  `Monitoring_Time` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  `Total_Phosphorus` float NULL DEFAULT NULL,
  `Algal_Density` float NULL DEFAULT NULL,
  `Conductivity` float NULL DEFAULT NULL,
  `Total_Nitrogen` float NULL DEFAULT NULL,
  `PH` float NULL DEFAULT NULL,
  `Section_Point` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  `Ammonia_Nitrogen` float NULL DEFAULT NULL,
  `Water_Quality_Type` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  `Permanganate_Index` float NULL DEFAULT NULL,
  `Site_Condition` varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL,
  PRIMARY KEY (`Mid`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = latin1 COLLATE = latin1_swedish_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;

 */

// 构造 SQL 语句和参数
//     println("SewageDischarge_Data_Process启动完成666")
//    WaterQualities.writeStream
//      .foreach(new BaseJdbcSink("replace into water_quality (Water_Temperature, Turbidity, Chlorophyll_Alpha, Watershed_River, Province_City, Dissolved_Oxygen, Monitoring_Time, Total_Phosphorus, Algal_Density, Conductivity, Total_Nitrogen, PH, Section_Point, Ammonia_Nitrogen, Water_Quality_Type, Permanganate_Index, Site_Condition) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") {
//        override def realProcess(sql: String, bean: WaterQualityBean): Unit = {
//
//          println("SewageDischarge_Data_Process启动完成777")
//
//val ps = conn.prepareStatement(sql)
//          println("SewageDischarge_Data_Process启动完成111")
//          ps.setFloat(1, bean.waterTemperature)
//          ps.setFloat(2, bean.turbidity)
//          ps.setFloat(3, bean.chlorophyllAlpha)
//          ps.setString(4, bean.watershedRiver)
//          ps.setString(5, bean.provinceCity)
//          ps.setFloat(6, bean.dissolvedOxygen)
//          ps.setString(7, bean.monitoringTime)
//          ps.setFloat(8, bean.totalPhosphorus)
//          ps.setFloat(9, bean.algalDensity)
//          ps.setFloat(10, bean.conductivity)
//          ps.setFloat(11, bean.totalNitrogen)
//          println("SewageDischarge_Data_Process启动完成111")
//          ps.setFloat(12, bean.pH)
//          println("SewageDischarge_Data_Process启动完成222")
//          ps.setString(13, bean.sectionPoint)
//          ps.setFloat(14, bean.ammoniaNitrogen)
//          ps.setString(15, bean.waterQualityType)
//          ps.setFloat(16, bean.permanganateIndex)
//          ps.setString(17, bean.siteCondition)
//          // 执行SQL语句
////          ps.executeUpdate()
////          val result = ps.executeUpdate()
//          val result = ps.executeUpdate()
//          println(s"Insert result: $result")
//        }
//      })
//      .outputMode("update")
//      .trigger(Trigger.ProcessingTime(0))
//      .option("truncate",false)
//      .start()
//        .awaitTermination()

        WaterQualities.writeStream
          .format("console")
          .outputMode("append")
          .trigger(Trigger.ProcessingTime(0))
          .option("truncate",false)
          .start()

    WaterQualities.writeStream
   .foreach(new BaseJdbcSink("replace into water_quality (Water_Temperature, Turbidity, Chlorophyll_Alpha, Watershed_River, Province_City, Dissolved_Oxygen, Monitoring_Time, Total_Phosphorus, Algal_Density, Conductivity, Total_Nitrogen, PH, Section_Point, Ammonia_Nitrogen, Water_Quality_Type, Permanganate_Index, Site_Condition) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)") {
     override def realProcess(sql: String, value: WaterQualityBean): Unit = {
          println("SewageDischarge_Data_Process启动完成666")
          ps = conn.prepareStatement(sql)
          ps.setFloat(1, value.Water_Temperature)
          ps.setFloat(2, value.Turbidity)
          ps.setFloat(3, value.Chlorophyll_Alpha)
          ps.setString(4, value.Watershed_River)
          ps.setString(5, value.Province_City)
          ps.setFloat(6, value.Dissolved_Oxygen)
          ps.setString(7, value.Monitoring_Time)
          ps.setFloat(8, value.Total_Phosphorus)
          ps.setFloat(9, value.Algal_Density)
          ps.setFloat(10, value.Conductivity)
          ps.setFloat(11, value.Total_Nitrogen)
          ps.setFloat(12, value.PH)
          ps.setString(13, value.Section_Point)
          ps.setFloat(14, value.Ammonia_Nitrogen)
          ps.setString(15, value.Water_Quality_Type)
          ps.setFloat(16, value.Permanganate_Index)
          ps.setString(17, value.Site_Condition)
          ps.executeUpdate()
        }
      })
     .outputMode("update")
           .trigger(Trigger.ProcessingTime(0))
           .option("truncate",false)
           .start()
             .awaitTermination()



//    WaterQualities.writeStream
//      .format("console")
//      .outputMode("append")
//      .trigger(Trigger.ProcessingTime(0))
//      .option("truncate",false)
////      .start()
//      .foreach(new BaseJdbcSink("replace into water_quality (Water_Temperature, Turbidity, Chlorophyll_Alpha, Watershed_River, Province_City, Dissolved_Oxygen, Monitoring_Time, Total_Phosphorus, Algal_Density, Conductivity, Total_Nitrogen, PH, Section_Point, Ammonia_Nitrogen, Water_Quality_Type, Permanganate_Index, Site_Condition) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)") {
//        override def realProcess(sql: String, value: WaterQualityBean): Unit = {
//          println("SewageDischarge_Data_Process启动完成666")
//          ps = coon.prepareStatement(sql)
////          var ps: PreparedStatement = null
////          var conn: Connection = null
//          try {
////            conn = getConnection()
//            ps = conn.prepareStatement(sql)
//            ps.setFloat(1, value.Water_Temperature)
//            ps.setFloat(2, value.Turbidity)
//            ps.setFloat(3, value.Chlorophyll_Alpha)
//            ps.setString(4, value.Watershed_River)
//            ps.setString(5, value.Province_City)
//            ps.setFloat(6, value.Dissolved_Oxygen)
//            ps.setString(7, value.Monitoring_Time)
//            ps.setFloat(8, value.Total_Phosphorus)
//            ps.setFloat(9, value.Algal_Density)
//            ps.setFloat(10, value.Conductivity)
//            ps.setFloat(11, value.Total_Nitrogen)
//            ps.setFloat(12, value.PH)
//            ps.setString(13, value.Section_Point)
//            ps.setFloat(14, value.Ammonia_Nitrogen)
//            ps.setString(15, value.Water_Quality_Type)
//            ps.setFloat(16, value.Permanganate_Index)
//            ps.setString(17, value.Site_Condition)
//            ps.executeUpdate()
//          } catch {
//            case e: Exception => e.printStackTrace()
//          } finally {
//            close(conn, ps)
//          }
//        }
//      })
//      .outputMode("update")
//      .trigger(Trigger.ProcessingTime(0))
//      .option("truncate",false)
//      .start()
//      .awaitTermination()




    println("SewageDischarge_Data_Process启动完成")

  }
}
